package com.bw.fun;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple9;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

/*
//map => （day,city,下单金额，下单次数，0，0，0,0.0,0.0）   order_info
        //map => （day,city,0，0，完成运输次数，完成里程总数，完成总时长,0.0,0.0）   transport_task
 */
public class Mn6AsyncIOFunc2 extends RichAsyncFunction<String, Tuple9<String,String,Double,Integer,Integer,Double,Long,Double,Double>> {
    Connection conn = null;
    PreparedStatement ps2 = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop102:2181");
        ps2 = conn.prepareStatement("select * from dim_base_region_info_yk9 where id=?" ); //行政区域表
        conn.setAutoCommit(true);
    }
    @Override
    public void close() throws Exception {
        ps2.close();
        conn.close();
    }

    @Override
    public void asyncInvoke(String v, ResultFuture<Tuple9<String,String,Double,Integer,Integer,Double,Long,Double,Double>> resultFuture) throws Exception {

        JSONObject data = JSON.parseObject(v).getJSONObject("data");
        // 发送者城市ID
        String city_id = data.getString("sender_city_id");
        ps2.setString(1,city_id);
        String cityName = null;
        ResultSet rs = ps2.executeQuery();
        if(rs.next()){
            String line = rs.getString("line");
            cityName = JSON.parseObject(line).getString("name");
        }
        // 日期截取
        String day = data.getString("create_time").substring(0, 10);
        //
        Double amount = data.getDouble("amount");

        //（day,city,下单金额，下单次数，0，0，0,0.0,0.0）   order_info
        resultFuture.complete(Collections.singleton(new Tuple9<>(day,cityName,amount,1,0,0.0,0L,0.0,0.0)));
    }
}
